package avicit.bdp.dgs.qa.service;

import avicit.bdp.common.base.BaseService;
import avicit.bdp.common.dto.ProcessInstanceDto;
import avicit.bdp.common.dto.query.dds.ExecutorQuery;
import avicit.bdp.common.dto.request.dds.ScheduleParamDto;
import avicit.bdp.common.service.dao.BdpCommonDao;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.BdpCommonService;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.*;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.dds.api.ExecutorApi;
import avicit.bdp.dds.api.ProcessDefinitionApi;
import avicit.bdp.dds.api.SchedulerApi;
import avicit.bdp.dds.dispatch.enums.*;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dgs.qa.dao.JobDAO;
import avicit.bdp.dgs.qa.dto.*;
import avicit.bdp.dgs.qa.service.algo.AlgoStrategySelector;
import avicit.bdp.dgs.qa.service.engine.EngineStrategySelector;
import avicit.bdp.dgs.qa.utils.constant.QaConstant;
import avicit.bdp.dgs.qa.utils.enums.*;
import avicit.bdp.dgs.qa.utils.redis.RedisUtils;
import avicit.platform6.api.sysmsg.dto.SysMsgDTO;
import avicit.platform6.api.system.ConvertColumnClient;
import avicit.platform6.api.system.SysMsgClient;
import avicit.platform6.api.system.impl.SystemConstant;
import avicit.platform6.commons.utils.BusinessUtil;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.commons.utils.JsonHelper;
import avicit.platform6.core.context.ThreadContextHelper;
import avicit.platform6.core.exception.BusinessException;
import avicit.platform6.core.properties.PlatformConstant;
import avicit.platform6.core.rest.msg.Muti3Bean;
import avicit.platform6.core.rest.msg.QueryRespBean;
import avicit.platform6.core.rest.msg.ResponseMsg;
import com.alibaba.fastjson2.JSONObject;
import com.github.pagehelper.Page;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.jetty.http.HttpStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @金航数码科技有限责任公司
 * @作者：developer
 * @邮箱：developer@avic-digital.com
 * @创建时间： 2023-03-30
 * @类说明：JobService
 * @修改记录：
 * @注意事项：
 * @主要功能：质量作业Service
 */
@Service
public class JobService extends BaseService<JobDAO, JobDTO> {
    private static final Logger logger = LoggerFactory.getLogger(JobService.class);

    @Autowired
    private RuleService ruleService;

    @Autowired
    private SchedulerService schedulerService;

    @Autowired
    private SubscribeService subscribeService;

    @Autowired
    private JobResultService jobResultService;

    @Autowired
    private JobReportService jobReportService;

    @Autowired
    private ConvertColumnClient convertColumnClient;

    @Autowired
    private DataSourceService commDsService;

    @Autowired
    private QaDataSourceService qaDsService;

    @Autowired
    private ProcessDefinitionApi processDefinitionApi;
    @Autowired
    private SchedulerApi schedulerApi;
    @Autowired
    private ExecutorApi executorApi;

    @Autowired
    private BdpCommonService bdpCommonService;

    @Resource
    private BdpCommonDao commonDao;

    @Resource
    RedisUtils redisUtils;

    @Autowired
    private SysMsgClient sysMsgClient;

    /**
     * 任务计算引擎
     */
    @Autowired
    private EngineStrategySelector engineStrategySelector;

    /**
     * 文件校验算法策略选择器
     */
    @Autowired
    private AlgoStrategySelector algoStrategySelector;

    /**
     * 新增质量作业
     *
     * @param jobDTO
     * @return
     */
    @Transactional
    public JobDTO addJob(JobDTO jobDTO) {
        checkAddValid(jobDTO);

        // 创建dds任务
        String processDefinitionId = createDdsTaskInner(jobDTO, new HashMap<>());

        // 设置作业主键ID
        jobDTO.setId(processDefinitionId);

        // 保存质量作业Job
        save2Db(jobDTO);

        // 文件校验任务，需要判断是否自动匹配模板
        autoCreateRule(jobDTO);

        // 持久化数据源信息
        saveDs2Db(jobDTO);

        // 新建作业默认为手动执行
        saveScheduleConf(jobDTO, SchedulerType.MANUAL);
        BdpLogUtil.log4Insert(jobDTO);
        return jobDTO;
    }

    /**
     * 新增质量作业
     *
     * @param map
     * @return
     */
    @Transactional
    public ResponseMsg<String> autoCreateFileJob(Map<String, Object> map) {
        Date startTime = new Date();

        // step1:合法性检查
        checkAutoCreateValid(map);

        // step2:map参数转为JobDTO
        JobDTO jobDTO = map2Job(map);

        // step3:设置主键ID
        jobDTO.setId(ComUtil.getId());

        // step4:自动匹配规则模板，生成规则
        List<RuleDTO> ruleList = this.ruleService.autoCreateRule(jobDTO);

        // step5:设置调度方式
        saveScheduleConf(jobDTO, SchedulerType.AUTO);

        // step6:持久化作业
        save2Db(jobDTO);

        // step7:立即执行文件校验，并返回执行结果
        ResponseMsg<List<FtdpFileDTO>> result = executeFileTaskImmediately(jobDTO, ruleList);

        // step8:计算执行时间
        String execTime = getExecTime(startTime, new Date());

        // step9:设置任务状态、日志信息、质量报告
        setFileJobState(result.getRetCode(), jobDTO);
        setFileJobLog(result, execTime, jobDTO);
        setFileJobReport(jobDTO, ruleList, result, execTime);

        // step10:更新任务
        update2Db(jobDTO);

        // step11:返回执行结果
        return getResMsg(result, jobDTO.getId());
    }

    /**
     * 删除/批量删除质量作业，多个id使用英文逗号隔开
     *
     * @param id
     * @return
     */
    @Transactional
    public int deleteJob(String id) {
        if (StringUtils.isBlank(id)) {
            return 0;
        }

        for (String jobId : Arrays.asList(id.trim().split(Constants.COMMA))) {
            // 删除数据源
            this.qaDsService.deleteDsByCond(id, null);

            // 删除规则
            this.ruleService.deleteRuleByJobId(jobId);

            // 删除调度
            this.schedulerService.deleteSchedulerByJobId(jobId);

            // 删除订阅
            this.subscribeService.deleteSubscribeByJobId(jobId);

            // 删除任务执行结果
            this.jobResultService.deleteResultByJobId(jobId);

            // 删除质量报告
            this.jobReportService.deleteJobReport(jobId, null);

            // 删除dds任务
            this.deleteDdsTaskInner(jobId);

            // 删除质量作业
            this.mapper.deleteJobByCond(Arrays.asList(jobId));
        }

        String logTitle = "删除质量任务：【" + id + "】";
        BdpLogUtil.log("质量任务模块", logTitle, PlatformConstant.OpType.delete);
        return 0;
    }

    /**
     * 修改质量作业
     *
     * @param jobDTO
     * @return
     */
    @Transactional
    public boolean updateJob(JobDTO jobDTO) {
        checkUpdateValid(jobDTO);

        update2Db(jobDTO);

        return true;
    }

    /**
     * 更新质量作业状态
     *
     * @param bean，第一个参数：jobId（processDefinitonId），第二个参数：processInstanceId，第三个参数：status
     * @return
     */
    @Transactional
    public boolean updateJobStatus(Muti3Bean<String, String, Integer> bean) {
        if (bean == null) {
            logger.error("dds传入参数为空，更新质量作业状态失败");
            return false;
        }

        // 更新任务状态
        JobDTO jobDTO = updateState2Db(bean);

        // 生成工单
        pushWorkOrder2Redis(jobDTO);

        // 生成质量报告
        this.jobReportService.pushJobReportCommand2Redis(jobDTO);

        // 根据订阅配置，通知任务执行结果
        /**
         * modified by wm,2023-12-19,生成质量报告时，有告警才发送系统消息。
         */
//        sendSysMsg(jobDTO);

        return true;
    }

    /**
     * 查询质量作业详情
     *
     * @param id 质量作业ID
     * @return
     */
    @Transactional
    public JobDTO getJobDetail(String id) {
        if (StringUtils.isBlank(id)) {
            return null;
        }

        JobDTO baseJob = this.selectByPrimaryKey(id);
        if (baseJob == null) {
            return null;
        }

        JobDTO queryResult = null;
        if (isStructDataMeasure(baseJob.getMeasureType())) {
            // 查找结构化数据校验作业类型
            queryResult = this.mapper.getJobDetail(id);
        } else {
            // 查找文件校验作业类型
            queryResult = this.selectByPrimaryKey(id);
        }

        queryResult.setSecretLevel(baseJob.getSecretLevel());
        // 转换
        if (queryResult != null) {
            valueConvert(Arrays.asList(queryResult));
        }

        return queryResult;
    }

    /**
     * 分页查询质量作业
     *
     * @param keyWords
     * @param measureType
     * @param schedulerType
     * @param resultType
     * @param state
     * @param datasourceName
     * @param tableName
     * @param pageNo
     * @param pageSize
     * @return
     */
    @Transactional
    public QueryRespBean<JobDTO> queryJob(String keyWords, Integer measureType, Integer schedulerType,
                                          Integer resultType, Integer state, String datasourceName,
                                          String tableName, Integer pageNo, Integer pageSize) {
        QueryRespBean<JobDTO> queryRespBean = new QueryRespBean<>();

        // step2：分页查询数据权限列表
        List<String> wordSecretList = ThreadContextHelper.getWordSecretList();
        com.github.pagehelper.PageHelper.startPage(pageNo, pageSize);
        Page<JobDTO> queryRet = null;
        if (measureType == MeasureType.FILE.getCode()) {
            // 查找文件校验作业类型
            queryRet = this.mapper.queryFileJobByCond(keyWords, measureType, schedulerType, resultType, state, wordSecretList);
        } else {
            // 查找结构化数据校验作业类型
            queryRet = this.mapper.queryJobByCond(keyWords, measureType, schedulerType, resultType, state, datasourceName, tableName, wordSecretList);
        }
        queryRespBean.setResult(queryRet);

        // step3；转换
        valueConvert(queryRet.getResult());
        BdpLogUtil.log4Query(queryRespBean);
        return queryRespBean;
    }

    /**
     * 判读模板名称是否重复
     *
     * @param jobId
     * @param name
     * @param measureType
     * @return true:重复；false:不重复
     */
    @Transactional
    public boolean isSameName(String jobId, String name, Integer measureType) {
        long count = this.mapper.selectNameCount(jobId, name, measureType);
        if (count > 0) {
            return true;
        }

        return false;
    }

    /**
     * 启动/停止/重新运行质量作业
     *
     * @param id                作业ID，多个ID使用英文逗号隔开
     * @param processInstanceId 流程实例ID
     * @param ruleIdList        规则ID列表（可选参数，支持两种启动方式：1、作业启动；2、指定作业下的规则启动）
     * @param mode              执行模式，0：启动、1：停止、2：重新运行
     * @param userOptions       用户配置参数
     * @return
     */
    @Transactional
    @SuppressWarnings("此处应使用分布式锁，集群部署时，保证状态一致性，待优化")
    public synchronized String executeJob(String id, String processInstanceId, List<String> ruleIdList, Integer mode, Map<String, Object> userOptions) {
        if (mode == null) {
            throw new BusinessException("执行模式为空");
        } else {
            if (mode < 0 || mode > 3) {
                throw new BusinessException("执行模式取值范围[0,2]，当前值：" + mode);
            }
        }
        if (StringUtils.isEmpty(id)) {
            throw new BusinessException("质量作业ID为空，启动/停止失败." + mode);
        }

        JobDTO job = this.getJobDetail(id);
        if (job == null) {
            throw new BusinessException("未找到质量作业,执行或停止质量作业失败,jobId=" + id);
        }

        // 命令与状态检查
        checkJobState(mode, job);

        // 更新dds任务
        updateDdsTaskInner(this.getJobDetail(id), new HashMap<>());

        // 启动dds任务
        String result = null;
        if (mode == 0) {
            job.setState(JobState.RUNNING.getCode());
            result = startDdsTaskInner(id);
        } else if (mode == 1) {
            // 停止dds任务
            job.setState(JobState.STOP.getCode());
            result = stopDdsTaskInner(id);
        } else if (mode == 2) {
            // 重新运行dds任务
            job.setState(JobState.RUNNING.getCode());
            result = restartDdsTaskInner(processInstanceId);
        }

        updateJob(job);

        String logTitle = "启动质量任务：【" + job.getName() + "】";
        BdpLogUtil.log("质量任务模块", logTitle, PlatformConstant.OpType.update);
        return result;
    }

    /**
     * 启用定时/停用定时
     *
     * @param jobId        作业ID，多个ID使用英文逗号隔开
     * @param releaseState 0：停用定时、1：启用定时
     * @return
     */
    @Transactional
    @SuppressWarnings("此处应使用分布式锁，集群部署时，保证状态一致性，待优化")
    public synchronized String releaseJob(String jobId, Integer releaseState) {
        if (releaseState == null) {
            throw new BusinessException("执行模式为空");
        } else {
            if (releaseState < 0 || releaseState > 1) {
                throw new BusinessException("执行模式取值范围[0,1]，当前值：" + releaseState);
            }
        }
        if (StringUtils.isEmpty(jobId)) {
            throw new BusinessException("质量作业ID为空，启用/停用定时失败." + releaseState);
        }

        JobDTO job = this.getJobDetail(jobId);
        if (job == null) {
            throw new BusinessException("未找到质量作业,启用/停用定时失败,jobId=" + jobId);
        }

        // 启用定时时，先更新DDS任务；停用定时时，不需要更新DDS任务；
        if (releaseState == 1) {
            updateDdsTaskInner(job, new HashMap<>());
        }

        // 启用/停用定时
        String result = releaseDdsTaskInner(jobId, releaseState);

        // 更新任务是否启用定时标记
        job.setEnableCrontab(releaseState); // 0:未启用定时、1：启用定时
        updateJob(job);

        return result;
    }

    /**
     * 查看日志
     *
     * @param jobId             根据作业ID查询最近一次执行日志
     * @param processInstanceId 根据流程实例ID查询历史任务执行日志
     * @param limit             跳过行数
     * @param skipLineNum       查询的行数
     * @return
     */
    @Transactional
    public String queryLog(String jobId, String processInstanceId, Integer skipLineNum, Integer limit) {
        if (StringUtils.isAllEmpty(jobId, processInstanceId)) {
            throw new BusinessException("作业ID和流程实例ID不能同时为空");
        }

        // 文件校验（自动执行）日志特殊处理
        JobDTO jobDetail = this.getJobDetail(jobId);
        if (jobDetail == null) {
            return null;
        }

        String logTitle = "查看：【" + jobDetail.getName() + "】的质量任务日志";
        BdpLogUtil.log("质量任务模块", logTitle, PlatformConstant.OpType.select);

        if (MeasureType.FILE.getCode() == jobDetail.getMeasureType() && isAutoExecJob(jobDetail.getSchedulerType())) {
            return jobDetail.getLog();
        }

        /**
         * 质量任务，直接查看日志时，仅传入jobId
         */
        if (StringUtils.isEmpty(processInstanceId)) {
            return this.bdpCommonService.showLog(processInstanceId, jobId, skipLineNum, limit);
        }

        /**
         * 查看历史任务日志时，传入jobId、processInstanceId
         */
        return this.bdpCommonService.showLog(processInstanceId, skipLineNum, limit);
    }

    /**
     * 查看质量报告
     *
     * @param jobId             根据质量作业ID查询最近一次执行结果
     * @param processInstanceId 根据流程实例ID查询历史任务执行结果
     * @return
     */
    @Transactional
    public Map<String, Object> queryReport(String jobId, String processInstanceId) {
        // 作业ID和流程实例ID不能同时为空
        if (StringUtils.isAllEmpty(jobId, processInstanceId)) {
            logger.error("作业ID和流程实例ID不能同时为空");
            return new HashMap<>();
        }

        // 作业是否存在
        JobDTO jobDetail = this.getJobDetail(jobId);
        if (jobDetail == null) {
            logger.error("未找到作业信息，jobId：" + jobId + "，procInsId：" + processInstanceId);
            return new HashMap<>();
        }

        String logTitle = "查看：【" + jobDetail.getName() + "】的质量报告";
        BdpLogUtil.log("质量任务模块", logTitle, PlatformConstant.OpType.select);

        // 获取质量报告
        JobReportDTO jobReport = this.jobReportService.queryJobReport(jobDetail, processInstanceId);
        if (jobReport != null) {
            return jobReport.getReportMap();
        }

        return new HashMap<>();
    }

    /**
     * 下载质量报告
     *
     * @param jobId             根据质量作业ID下载最近一次执行结果
     * @param processInstanceId 根据流程实例ID下载历史任务执行结果
     * @param response
     * @return
     */
    @Transactional
    public void downloadReport(String jobId, String processInstanceId, HttpServletResponse response) {
        if (StringUtils.isAllEmpty(jobId, processInstanceId)) {
            throw new BusinessException("作业ID和流程实例ID不能同时为空");
        }

        JobDTO jobDetail = this.getJobDetail(jobId);
        if (jobDetail == null) {
            throw new BusinessException("未找到作业信息，jobId：" + jobId + "，procInsId：" + processInstanceId);
        }
        this.jobReportService.downloadReport(jobDetail, processInstanceId, response);
    }

    /**
     * 查看脏数据
     *
     * @param jobId             根据质量作业ID查看最近一次执行的脏数据
     * @param processInstanceId 根据流程实例ID查看历史任务脏数据
     * @param ruleId            规则ID
     * @return
     */
    @Transactional
    public Map<String, Object> queryDirtyData(String jobId, String processInstanceId, String ruleId) {
        if (StringUtils.isBlank(jobId) && StringUtils.isBlank(processInstanceId)) {
            throw new BusinessException("作业ID和流程实例ID不能同时为空");
        }

        throw new BusinessException("暂不支持该接口，敬请期待!");
    }

    /**
     * 查询历史任务
     *
     * @param id
     * @param pageNo
     * @param pageSize
     * @return
     */
    @Transactional
    public QueryRespBean<JobDTO> queryHistoryTask(String id, Integer pageNo, Integer pageSize) {
        if (StringUtils.isBlank(id)) {
            throw new BusinessException("作业ID为空");
        }

        // step1:根据jobId查询流程实例
        QueryRespBean queryRespBean = new QueryRespBean();
        com.github.pagehelper.PageHelper.startPage(pageNo, pageSize);
        Page<ProcessInstanceDto> queryRet = this.commonDao.queryProcessInstanceByProjectId(null, id, null,
                null, null, null, new ArrayList<>(), null);
        if (CollectionUtils.isEmpty(queryRet)) {
            return queryRespBean;
        }

        // step2:流程实例转为JobDTO
        List<JobDTO> jobList = instanceConvertJob(id, queryRet.getResult());
        Page<JobDTO> page = new Page(pageNo, pageSize);
        page.addAll(jobList);
        page.setPages(queryRet.getPages());
        page.setTotal(queryRet.getTotal());
        queryRespBean.setResult(page);

        BdpLogUtil.log4Query("质量任务模块", "历史任务", queryRespBean);
        return queryRespBean;
    }

    /**
     * 查询监控任务(所有质量任务)
     *
     * @param projectId
     * @param jobId
     * @param jobName
     * @param startTime
     * @param endTime
     * @param state
     * @param pageNo
     * @param pageSize
     * @return
     */
    @Transactional
    public QueryRespBean<JobDTO> queryMonitorTask(String projectId, String jobId, String jobName,
                                                  String startTime, String endTime, Integer state,
                                                  Integer pageNo, Integer pageSize) {
        // step1:设置默认项目空间
        if (StringUtils.isEmpty(projectId)) {
            projectId = QaConstant.DEFAULT_PROJECT_ID;
        }
        List<Integer> stateList = new ArrayList<>();
        if (state != null && state != -1) {
            // -1表示全部状态
            stateList.add(state);
        }

        // step2:根据项目ID分页查询流程实例
        List<String> wordSecretList = ThreadContextHelper.getWordSecretList();
        com.github.pagehelper.PageHelper.startPage(pageNo, pageSize);
        Page<ProcessInstanceDto> procInsList = this.commonDao.queryProcessInstanceByProjectId(projectId, jobId, jobName,
                startTime, endTime, null, stateList, wordSecretList);
        if (CollectionUtils.isEmpty(procInsList)) {
            logger.debug("未找到监控任务,projectId=" + projectId + ",jobId=" + jobId + ",jobName=" + jobName + ",state=" + state);
            return new QueryRespBean();
        }

        // step3:流程实例转化为作业
        List<JobDTO> jobList = instanceConvertJob(null, procInsList.getResult());
        covertUserName(jobList);

        QueryRespBean queryRespBean = new QueryRespBean();
        Page<JobDTO> page = new Page(pageNo, pageSize);
        page.addAll(jobList);
        page.setPages(procInsList.getPages());
        page.setTotal(procInsList.getTotal());
        queryRespBean.setResult(page);

        BdpLogUtil.log4Query("质量任务模块", "质量任务", queryRespBean);
        return queryRespBean;
    }

    /**
     * 获取质量作业数量
     *
     * @param startTime
     * @param endTime
     * @return
     */
    @Transactional
    public Long countJob(String startTime, String endTime, List<String> wordSecretList) {
        return this.mapper.countJob(startTime, endTime, wordSecretList);
    }

    /**
     * 统计评估次数
     *
     * @param startTime
     * @param endTime
     * @return
     */
    @Transactional
    public Long countProcessInstance(String startTime, String endTime) {
        return this.commonDao.countProcessInstance(QaConstant.DEFAULT_PROJECT_ID, startTime, endTime);
    }

    /**
     * 删除/批量删除质量作业，多个id使用英文逗号隔开
     *
     * @param id
     * @return
     */
    @Transactional
    public int deleteMonitorTask(String id) {
        if (StringUtils.isBlank(id)) {
            return 0;
        }

        List<String> insIdList = Arrays.asList(id.trim().split(Constants.COMMA));
        if (CollectionUtils.isEmpty(insIdList)) {
            return 0;
        }

        for (String insId : insIdList) {
            // 删除执行结果bdp_dgs_qa_job_result
            this.jobResultService.deleteResultByInstanceId(id);

            // 删除质量报告
            this.jobReportService.deleteJobReport(null, insIdList);

            // 删除流程实例
            this.commonDao.deleteProcessInstanceById(insId);
        }
        String logTitle = "删除质量监控：【" + id + "】";
        BdpLogUtil.log("质量监控模块", logTitle, PlatformConstant.OpType.delete);
        return insIdList.size();
    }

    /**
     * 获取流程实例状态
     *
     * @param jobId
     * @param processInstanceId
     * @return
     */
    public Integer getProcessInstanceState(String jobId, String processInstanceId) {
        ProcessInstanceDto ins = this.getProcessInstance(jobId, processInstanceId);
        if (ins == null) {
            return -1;
        }

        logger.warn(String.format("流程定义：%s, 流程实例：%s，状态：%s", ins.getProcessDefinitionId(), ins.getId(), ins.getState()));

        return ins.getState();
    }

    /**
     * 根据作业状态统计数目
     *
     * @param startTime
     * @param endTime
     * @return
     */
    public List<Map<String, Object>> queryJobStateDistribution(String startTime, String endTime) {
        List<Map<String, Object>> resultList = new ArrayList<>();
        for (JobState state : JobState.values()) {
            long jobNum = this.mapper.countJobByState(state.getCode(), startTime, endTime);
            Map<String, Object> map = new HashMap<>();
            map.put("name", state.getDesc());
            map.put("value", jobNum);
            resultList.add(map);
        }

        return resultList;
    }

    /**
     * 新建作业时合法性检查
     *
     * @param jobDTO
     */
    private void checkAddValid(JobDTO jobDTO) {
        if (jobDTO == null) {
            throw new BusinessException("入参为空.");
        }
        if (StringUtils.isBlank(jobDTO.getName())) {
            throw new BusinessException("作业名字为空.");
        }
        if (jobDTO.getMeasureType() == null) {
            throw new BusinessException("监控类型为空.");
        } else {
            if (MeasureType.getDescByCode(jobDTO.getMeasureType()) == null) {
                throw new BusinessException("监控类型不合法,当前值：" + jobDTO.getMeasureType() + ",取值范围[0,6]");
            }
            if (isStructDataMeasure(jobDTO.getMeasureType())) {
                if (StringUtils.isBlank(jobDTO.getDatasourceId())) {
                    throw new BusinessException("数据源ID为空.");
                }
                if (StringUtils.isBlank(jobDTO.getDbName())) {
                    throw new BusinessException("库名为空.");
                }
            }
        }
    }

    /**
     * 试飞文件上传时自动创建质量作业
     *
     * @param map
     */
    private void checkAutoCreateValid(Map<String, Object> map) {
        if (map == null) {
            throw new BusinessException("参数为空.");
        }
        if (map.size() <= 0) {
            throw new BusinessException("未配置参数.");
        }
        if (map.get(QaConstant.FILE_TYPE) == null) {
            throw new BusinessException("fileType字段为空，未配置试飞文件类型（0：试飞原始数据、1：工程值数据）.");
        }
        if (map.get(QaConstant.FTDP_OFFSITE_DATA_ID) == null) {
            throw new BusinessException("ftdpOffsiteDataId字段为空，未配置试飞信息主键ID.");
        }
    }

    /**
     * 更新作业时合法性检查
     *
     * @param jobDTO
     */
    private void checkUpdateValid(JobDTO jobDTO) {
        if (jobDTO == null) {
            throw new BusinessException("入参为空.");
        }
        if (StringUtils.isBlank(jobDTO.getId())) {
            throw new BusinessException("作业主键ID为空.");
        }
    }

    /**
     * 持久化DB
     *
     * @param jobDTO
     */
    private void save2Db(JobDTO jobDTO) {
        if (jobDTO == null) {
            return;
        }

        // 文件校验时，试飞信息存放在attr字段中
        if (jobDTO.getMeasureType().equals(MeasureType.FILE.getCode())) {
            jobDTO.setAttr(JSONObject.toJSONString(covert2OffsiteData(jobDTO)));
        }

        // 设置作业状态，0:草稿、1:未启动、2:运行中、3:成功、4:失败
        jobDTO.setState(JobState.EDIT.getCode());

        // 持久化
        this.insert(jobDTO);
    }

    /**
     * 持久化作业关联数据源到DB
     *
     * @param jobDTO
     */
    private void saveDs2Db(JobDTO jobDTO) {
        if (jobDTO == null) {
            return;
        }

        // 结构化数据采集作业，保存数据源信息
        if (isStructDataMeasure(jobDTO.getMeasureType())) {
            DataSourceDTO dataSource = this.commDsService.queryDataSourceById(jobDTO.getDatasourceId());
            if (dataSource == null) {
                throw new BusinessException("未找到指定数据源,创建或更新质量任务失败,jobId=" + jobDTO.getId());
            }

            List<QaDataSourceDTO> dsList = new ArrayList<>();
            QaDataSourceDTO ds = new QaDataSourceDTO();
            ds.setId(ComUtil.getId());
            ds.setBusinessId(jobDTO.getId());
            ds.setDatasourceId(dataSource.getId());
            ds.setDatasourceName(dataSource.getDatasourceName());
            ds.setDbName(jobDTO.getDbName());
            ds.setTableName(jobDTO.getTableName());
            ds.setMeasureType(jobDTO.getMeasureType());

            dsList.add(ds);
            this.qaDsService.batchInsertDs(jobDTO.getId(), dsList);
        }
    }

    /**
     * 新增质量作业时，调度方式默认为手动执行
     *
     * @param jobDTO
     * @param schedulerType
     */
    private void saveScheduleConf(JobDTO jobDTO, SchedulerType schedulerType) {
        SchedulerDTO schedulerDTO = new SchedulerDTO();
        schedulerDTO.setJobId(jobDTO.getId());
        schedulerDTO.setType(schedulerType.getCode());
        this.schedulerService.saveScheduler(schedulerDTO);
    }

    /**
     * 更新到DB
     *
     * @param jobDTO
     */
    private void update2Db(JobDTO jobDTO) {
        if (jobDTO == null) {
            return;
        }

        JobDTO oldDto = this.selectByPrimaryKey(jobDTO.getId());
        if (oldDto == null) {
            throw new BusinessException("根据ID未找到作业，id=" + jobDTO.getId() + "，更新失败");
        }

        // 持久化
        jobDTO.setCreatedBy(ThreadContextHelper.getUserId());
        this.updateByPrimaryKeySelective(jobDTO);

        // 审计日志
        BdpLogUtil.log4Update(jobDTO, oldDto);
    }

    /**
     * 更新状态到DB
     *
     * @param bean
     */
    private JobDTO updateState2Db(Muti3Bean<String, String, Integer> bean) {
        JobDTO jobDTO = new JobDTO();
        jobDTO.setId(bean.getDto1());
        jobDTO.setProcessInstanceId(bean.getDto2());
        jobDTO.setState(bean.getDto3());

        logger.info("开始更新任务状态,jobId:{},processInstanceId:{},state:{}", bean.getDto1(), bean.getDto2(), bean.getDto3());

        JobDTO oldDto = this.selectByPrimaryKey(jobDTO.getId());
        if (oldDto == null) {
            throw new BusinessException("根据ID未找到作业，id=" + jobDTO.getId() + "，更新失败");
        }

        // 持久化
        jobDTO.setCreatedBy(oldDto.getCreatedBy());
        this.updateByPrimaryKeySelective(jobDTO);

        // 审计日志
        BdpLogUtil.log4Update(jobDTO, oldDto);

        jobDTO.setName(oldDto.getName());
        jobDTO.setMeasureType(oldDto.getMeasureType());
        return jobDTO;
    }

    /**
     * 通过平台API将字段值转换为名称，包括通用选择组件、通用代码
     *
     * @param dtoList
     */
    private void valueConvert(List<JobDTO> dtoList) {
        if (CollectionUtils.isEmpty(dtoList)) {
            return;
        }

        // 设置创建者名称
        Map<String, Set<String>> convertFormData = new HashMap<>();
        for (JobDTO dto : dtoList) {
            BusinessUtil.createConvertSet(convertFormData, SystemConstant.USER, dto.getCreatedBy());
            BusinessUtil.createConvertSet(convertFormData, Constants.BDP_DATA_SECRET_LEVEL_KEY, dto.getSecretLevel());
        }
        if (convertFormData.size() > 0) {
            //获取请求结果
            Map<String, Map<String, String>> convertResultData = convertColumnClient.replace(convertFormData);
            //循环设置Alias或Name的值
            for (JobDTO dto : dtoList) {
                dto.setCreatedBy(BusinessUtil.convertFormat(convertResultData, SystemConstant.USER, dto.getCreatedBy()));
                dto.setSecretLevelName(BusinessUtil.convertFormat(
                        convertResultData, Constants.BDP_DATA_SECRET_LEVEL_KEY, dto.getSecretLevel()));
            }
        }

        // 更新任务状态
        handleJobState(dtoList);

        // 设置枚举类型
        for (JobDTO dto : dtoList) {
            dto.setMeasureTypeName(MeasureType.getDescByCode(dto.getMeasureType()));
            dto.setStateName(JobState.getDescByCode(dto.getState()));
            if (dto.getSchedulerType() != null) {
                dto.setSchedulerTypeName(SchedulerType.getDescByCode(dto.getSchedulerType()));
            }

            // 设置下次执行时间
            dto.setNextExecTime(getNextExecTime(dto));

            // 质量告警结果
            JobResultType resultType = getJobResultType(dto);
            if (resultType != null) {
                dto.setResultTypeName(resultType.getDesc());
            }

            // 设置表名称
            if (StringUtils.isEmpty(dto.getTableName())) {
                dto.setTableName(QaConstant.DOUBLE_HYPHEN);
            }

            // 设置文件校验信息
            setOffsiteData(dto);
        }
    }

    /**
     * 文件校验作业，设置试飞信息
     *
     * @param dto
     */
    private void setOffsiteData(JobDTO dto) {
        if (dto.getMeasureType() != MeasureType.FILE.getCode()) {
            return;
        }

        FtdpOffsiteDataDTO offsiteData = JSONObject.parseObject(dto.getAttr(), FtdpOffsiteDataDTO.class);
        if (offsiteData != null) {
            dto.setOffsiteDataId(offsiteData.getId());
            dto.setFileType(offsiteData.getFileType());
            dto.setFileTypeName(FileType.getDescByCode(offsiteData.getFileType()));
            dto.setTypeName(offsiteData.getTypeName());
            dto.setModelCode(offsiteData.getModelCode());
            dto.setAircraftName(offsiteData.getAircraftName());
            dto.setFlightDate(offsiteData.getFlightDate());
            dto.setSortie(offsiteData.getSortie());
            dto.setTestSystemName(offsiteData.getTestSystemName());
            dto.setDataProp(offsiteData.getDataProp());
            dto.setResourceIds(offsiteData.getResourceIds());
            dto.setLog(offsiteData.getLog());
        }
    }

    /**
     * userId -> userName
     *
     * @param dtoList
     */
    private void covertUserName(List<JobDTO> dtoList) {
        if (CollectionUtils.isEmpty(dtoList)) {
            return;
        }

        // 设置创建者名称
        Map<String, Set<String>> convertFormData = new HashMap<>();
        for (JobDTO dto : dtoList) {
            BusinessUtil.createConvertSet(convertFormData, SystemConstant.USER, dto.getCreatedBy());
        }
        if (convertFormData.size() > 0) {
            //获取请求结果
            Map<String, Map<String, String>> convertResultData = convertColumnClient.replace(convertFormData);
            //循环设置Alias或Name的值
            for (JobDTO dto : dtoList) {
                dto.setCreatedBy(BusinessUtil.convertFormat(convertResultData, SystemConstant.USER, dto.getCreatedBy()));
            }
        }
    }

    /**
     * 获取定时调度-下次执行时间
     *
     * @param jobDTO
     * @return
     */
    private String getNextExecTime(JobDTO jobDTO) {
        if (jobDTO == null) {
            return QaConstant.DOUBLE_HYPHEN;
        }
        if (jobDTO.getSchedulerType() != null) {
            // 手动执行，不需要设置执行时间
            if (SchedulerType.MANUAL.getCode() == jobDTO.getSchedulerType()) {
                return QaConstant.DOUBLE_HYPHEN;
            }
            // 自动执行，不需要设置执行时间
            if (isAutoExecJob(jobDTO.getSchedulerType())) {
                return QaConstant.DOUBLE_HYPHEN;
            }
            // 未启用定时，不需要设置下次执行时间
            if (jobDTO.getEnableCrontab() != null && jobDTO.getEnableCrontab() == 0) {
                return QaConstant.DOUBLE_HYPHEN;
            }
        }

        try {
            ScheduleParamDto scheduleParam = new ScheduleParamDto();
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat(Constants.YYYY_MM_DD_HH_MM_SS);
            Map<String, Object> schedulerConf = this.schedulerService.getScheduleDetail(jobDTO.getId());
            if (schedulerConf != null) {
                scheduleParam.setCrontab((String) schedulerConf.get("crontab"));
                Date startTime = simpleDateFormat.parse((String) schedulerConf.get("startTime"));
                Date endTime = simpleDateFormat.parse((String) schedulerConf.get("endTime"));
                scheduleParam.setStartTime(startTime);
                scheduleParam.setEndTime(endTime);
            }
            List<Date> dates = schedulerApi.previewSchedule(scheduleParam);
            if (CollectionUtils.isNotEmpty(dates)) {
                Date date = dates.get(0);
                return simpleDateFormat.format(date);
            }
        } catch (Exception e) {
            logger.error("获取下次执行时间失败,jobId=" + jobDTO.getId() + "errMsg=" + e.getMessage());
        }

        return QaConstant.DOUBLE_HYPHEN;
    }

    /**
     * 质量任务是否告警
     *
     * @param jobDTO
     */
    private JobResultType getJobResultType(JobDTO jobDTO) {
        if (jobDTO == null) {
            return null;
        }
        if (StringUtils.isAllEmpty(jobDTO.getId(), jobDTO.getProcessInstanceId())) {
            logger.debug("作业ID和流程实例ID同时为空，获取任务结果失败。");
            return null;
        }

        // 文件校验 && 自动执行，任务执行成功：正常；任务执行失败：告警
        if (MeasureType.FILE.getCode() == jobDTO.getMeasureType() && isAutoExecJob(jobDTO.getSchedulerType())) {
            if (JobState.SUCCESS.getCode() == jobDTO.getState()) {
                return JobResultType.NORMAL;
            } else {
                return JobResultType.FAIL;
            }
        }

        JobReportDTO jobReport = this.jobReportService.queryJobReport(jobDTO, jobDTO.getProcessInstanceId());
        if (jobReport == null || jobReport.getJobResultType() == null) {
            return JobResultType.FAIL;
        }

        return JobResultType.getTypeByCode(jobReport.getJobResultType());
    }

    /**
     * 创建dds任务
     *
     * @param job
     * @param userOptions
     * @return 流程定义ID
     */
    private String createDdsTaskInner(JobDTO job, Map<String, Object> userOptions) {
        // step1:填充ProcessDefinition
        Map<String, Object> defeJsonMap = constructProcDefe(job, userOptions);

        // step2:持久化流程定义
        return saveProcDefe(job, defeJsonMap);
    }

    /**
     * 更新dds任务
     *
     * @param job
     * @param userOptions
     */
    private void updateDdsTaskInner(JobDTO job, Map<String, Object> userOptions) {
        // step1:填充ProcessDefinition
        Map<String, Object> defeJsonMap = constructProcDefe(job, userOptions);

        // step2:持久化流程定义
        updateProcDefe(job, defeJsonMap);
    }

    /**
     * 删除dds任务
     *
     * @param jobId
     * @return
     */
    private boolean deleteDdsTaskInner(String jobId) {
        JobDTO jobDTO = this.selectByPrimaryKey(jobId);
        if (jobDTO != null) {
            // 文件校验-自动执行，没有创建DDS任务，不需要删除
            if (jobDTO.getSchedulerType() != null && isAutoExecJob(jobDTO.getSchedulerType())) {
                return true;
            }

            return this.processDefinitionApi.deleteProcessDefinition(jobId);
        }

        return true;
    }

    /**
     * 启动dds任务
     *
     * @param jobId
     * @return
     */
    private String startDdsTaskInner(String jobId) {
        ExecutorQuery executorQuery = new ExecutorQuery();
        executorQuery.setExecType(CommandType.START_PROCESS.getCode());
        executorQuery.setFailureStrategy(FailureStrategy.END.getCode());
        executorQuery.setProcessDefinitionId(jobId);
        executorQuery.setRunMode(RunMode.RUN_MODE_SERIAL.getCode());
        executorQuery.setTimeout(24 * 3600);
        executorQuery.setProcessInstancePriority(Priority.MEDIUM.getCode());
        executorQuery.setTaskDependType(TaskDependType.TASK_POST.getCode());// 一定要加dependType，否则rerun时dependType类型为null，抛出NullPointerException异常

        return this.executorApi.startProcess(executorQuery);
    }

    /**
     * 停止dds任务
     *
     * @param jobId
     * @return
     */
    private String stopDdsTaskInner(String jobId) {
        boolean result = this.executorApi.execute(jobId, "STOP");
        if (result) {
            return "停止成功。";
        }

        return "停止失败。";
    }

    /**
     * 重新运行dds任务
     *
     * @param processInstanceId
     * @return
     */
    private String restartDdsTaskInner(String processInstanceId) {
        boolean result = this.executorApi.executeByProcessInstanceId(processInstanceId, "REPEAT_RUNNING");
        if (result) {
            return "停止成功。";
        }

        return "停止失败。";
    }

    /**
     * dds任务上线/下线
     *
     * @param processDefinitonId
     * @param releaseState       0：offline；1：online
     * @return
     */
    private String releaseDdsTaskInner(String processDefinitonId, Integer releaseState) {
        // 0：offline；1：online
        boolean result = this.processDefinitionApi.releaseProcessDefinition(processDefinitonId, releaseState);
        if (result) {
            return releaseState == 0 ? "下线成功。" : "上线成功。";
        }

        return releaseState == 0 ? "下线失败。" : "上线失败。";
    }

    /**
     * 检查任务状态、启动命令
     *
     * @param mode
     * @param job
     */
    private void checkJobState(Integer mode, JobDTO job) {
        if (mode == null) {
            return;
        }

        JobState jobState = null;
        if (job.getState() != null) {
            jobState = JobState.getTypeByCode(job.getState());
        }
        if (jobState == null) {
            return;
        }

        // 启动/重新启动
        if (mode == 0 || mode == 2) {
            if (jobState == JobState.RUNNING) {
                throw new BusinessException("任务正在运行中，无法启动.");
            }
        } else if (mode == 1) {
            if (jobState == JobState.STOP || jobState == JobState.FAIL || jobState == JobState.SUCCESS) {
                throw new BusinessException("任务已经停止.");
            }
        }
    }

    /**
     * 创建dds中流程定义信息
     *
     * @param job
     * @param defeJsonMap
     * @return 流程定义ID
     */
    private String saveProcDefe(JobDTO job, Map<String, Object> defeJsonMap) {
        Map<String, Object> processDefeMap = buildProcDefeMap(job, defeJsonMap);

        return this.processDefinitionApi.createProcessDefinition(processDefeMap);
    }

    /**
     * 更新dds中流程定义信息
     *
     * @param job
     * @param defeJsonMap
     */
    private void updateProcDefe(JobDTO job, Map<String, Object> defeJsonMap) {
        Map<String, Object> processDefeMap = buildProcDefeMap(job, defeJsonMap);

        this.processDefinitionApi.updateProcessDefinition(processDefeMap);
    }

    /**
     * 流程定义Map
     *
     * @param job
     * @param defeJsonMap
     * @return
     */
    private Map<String, Object> buildProcDefeMap(JobDTO job, Map<String, Object> defeJsonMap) {
        Map<String, Object> processDefeMap = new HashMap<>();

        processDefeMap.put("id", job.getId());
        processDefeMap.put("name", job.getName());
        processDefeMap.put("releaseState", ReleaseState.OFFLINE.getCode());
        processDefeMap.put("projectId", QaConstant.DEFAULT_PROJECT_ID);
        processDefeMap.put("typeId", null);
        processDefeMap.put("type", 2);
        processDefeMap.put("remark", "质量作业...");
        processDefeMap.put("datasourceId", "commonFlow");
        processDefeMap.put("extraParams", null);
        processDefeMap.put("globalParams", JSONUtils.toJsonString(new ArrayList<>()));
        processDefeMap.put("processDefinitionJson", JSONUtils.toJsonString(defeJsonMap));
        processDefeMap.put("locations", null);
        processDefeMap.put("connects", JSONUtils.toJsonString(new ArrayList<>()));
        processDefeMap.put("secretLevel", job.getSecretLevel());

        return processDefeMap;
    }

    /**
     * 根据引擎类型，创建DDS任务
     *
     * @param job
     * @param userOptions
     * @return
     */
    @SuppressWarnings("注意：后续新增校验类型时，需要修改此处代码")
    private Map<String, Object> constructProcDefe(JobDTO job, Map<String, Object> userOptions) {
        String engineType = null;
        if (job.getMeasureType().equals(MeasureType.FILE.getCode())) {
            engineType = "file";
        } else {
            engineType = ConfigUtils.getInstance().getString("bdp.qa.task.engine", "sql");
        }

        return this.engineStrategySelector.getEngineByType(engineType).build(job, userOptions);
    }

    /**
     * 流程实例转换为JobDTO
     *
     * @param id
     * @param processInstanceList
     * @return
     */
    @SuppressWarnings("方法过长，待优化")
    private List<JobDTO> instanceConvertJob(String id, List<ProcessInstanceDto> processInstanceList) {
        List<JobDTO> jobList = new ArrayList<>();
        if (CollectionUtils.isEmpty(processInstanceList)) {
            return jobList;
        }

        for (ProcessInstanceDto ins : processInstanceList) {
            if (ins == null) {
                continue;
            }

            // 文件校验，自动执行时需要特殊处理
            JobDTO jobDetail = this.getJobDetail(ins.getProcessDefinitionId());
            if (jobDetail == null) {
                logger.error("未找到质量作业，jobId:" + ins.getProcessDefinitionId());
                continue;
            }

            JobDTO jobDTO = new JobDTO();
            jobDTO.setId(ins.getProcessDefinitionId());
            jobDTO.setName(ins.getName());
            jobDTO.setMeasureType(jobDetail.getMeasureType());
            jobDTO.setMeasureTypeName(jobDetail.getMeasureTypeName());

            // 任务状态处理
            if (ins.getState() != null) {
                JobState state = null;
                ExecutionStatus.of(ins.getState());
                switch (ins.getState()) {
                    case 0:
                        state = JobState.EDIT;
                        break;
                    case 1:
                        state = JobState.RUNNING;
                        break;
                    case 5:
                        state = JobState.STOP;
                        break;
                    case 6:
                        state = JobState.FAIL;
                        break;
                    case 7:
                        state = JobState.SUCCESS;
                        break;
                    default:
                        state = JobState.FAIL;
                        break;
                }

                jobDTO.setState(state.getCode());
                jobDTO.setStateName(state.getDesc());
            }

            // 调度类型处理
            if (ins.getScheduleTime() != null) {
                jobDTO.setSchedulerType(SchedulerType.PERIOD.getCode());
                jobDTO.setSchedulerTypeName(SchedulerType.PERIOD.getDesc());
            } else {
                if (!isAutoExecJob(jobDetail.getSchedulerType())) {
                    jobDTO.setSchedulerType(SchedulerType.MANUAL.getCode());
                    jobDTO.setSchedulerTypeName(SchedulerType.MANUAL.getDesc());
                } else {
                    // 文件校验 + 自动执行，需要特殊处理
                    jobDTO.setSchedulerType(SchedulerType.AUTO.getCode());
                    jobDTO.setSchedulerTypeName(SchedulerType.AUTO.getDesc());
                }
            }

            jobDTO.setResultType(null);
            jobDTO.setResultTypeName(null);
            jobDTO.setStartTime(ins.getStartTime());
            jobDTO.setEndTime(ins.getEndTime());
            jobDTO.setCreatedBy(ins.getCreatedBy());
            jobDTO.setProcessInstanceId(ins.getId());
            jobDTO.setSecretLevel(jobDetail.getSecretLevel());

            jobList.add(jobDTO);
        }

        return jobList;
    }

    /**
     * 获取流程实例
     *
     * @param jobId
     * @param processInstanceId
     * @return
     */
    private ProcessInstanceDto getProcessInstance(String jobId, String processInstanceId) {
        if (StringUtils.isBlank(jobId)) {
            logger.warn("质量作业ID为空,查询流程实例失败.");
            return null;
        }

        // 流程实例ID为空，查询最近一次执行结果
        if (StringUtils.isEmpty(processInstanceId)) {
            return this.commonDao.getLatestProcessInstanceByProcessDefinitionId(jobId);
        }

        // 流程实例ID不为空，需要匹配
        Page<ProcessInstanceDto> queryRet = this.commonDao.queryProcessInstanceListPaging(jobId, null, null, null, null, null, null);
        if (CollectionUtils.isEmpty(queryRet)) {
            logger.warn("未找到流程实例,jobId=" + jobId + ",instanceId=" + processInstanceId);
            return null;
        }

        for (ProcessInstanceDto ins : queryRet.getResult()) {
            if (ins == null || StringUtils.isEmpty(ins.getId())) {
                continue;
            }

            if (ins.getId().equals(processInstanceId)) {
                return ins;
            }
        }

        logger.warn("未找到流程实例,jobId=" + jobId + ",instanceId=" + processInstanceId);

        return null;
    }

    /**
     * 生成工单
     *
     * @param jobDTO
     */
    private void pushWorkOrder2Redis(JobDTO jobDTO) {
        if (jobDTO == null) {
            logger.warn("质量作业为空，生成工单失败.");
            return;
        }

        // 构建工单
        WorkOrderDTO workOrderDTO = new WorkOrderDTO();
        workOrderDTO.setName(jobDTO.getName());
        workOrderDTO.setLevel(WorkOrderLevel.MEDIUM.getCode()); // 默认工单等级为“中”
        workOrderDTO.setJobId(jobDTO.getId());
        workOrderDTO.setProcessInstanceId(jobDTO.getProcessInstanceId());
        workOrderDTO.setJobState(jobDTO.getState());

        // 向redis消息队列插入一条记录，后台自动生成工单
        redisUtils.lpush(QaConstant.WORK_ORDER_REDIS_KEY, JsonHelper.getInstance().writeValueAsString(workOrderDTO));

        logger.info("质量任务执行完毕，提交工单信息成功：\n" + JSONObject.toJSONString(workOrderDTO));
    }

    /**
     * 发送订阅消息
     *
     * @param jobDTO
     * @return
     */
    @SuppressWarnings("当前仅发送任务执行结果，后续会根据质量报警情况发送，2023-06-27，modified by wm")
    private boolean sendSysMsg(JobDTO jobDTO) {
        logger.info("当前任务：" + JSONObject.toJSONString(jobDTO));
        List<SubscribeDTO> subscribeList = this.subscribeService.getSubscribeList(jobDTO.getId());
        if (CollectionUtils.isEmpty(subscribeList)) {
            logger.warn("未找到订阅信息，jobId：" + jobDTO.getId());
            return true;
        }

        for (SubscribeDTO subscribe : subscribeList) {
            if (subscribe == null) {
                continue;
            }
            if (StringUtils.isEmpty(subscribe.getUserId())) {
                logger.warn("订阅用户ID为空.");
                continue;
            }

            SysMsgDTO msg = new SysMsgDTO();
            JobState jobState = JobState.getTypeByCode(jobDTO.getState());
            switch (jobState) {
                case FAIL:
                    msg.setTitle("质量作业【" + jobDTO.getName() + "】执行" + JobState.getDescByCode(jobDTO.getState()));
                    break;
                case STOP:
                    msg.setTitle("质量作业【" + jobDTO.getName() + "】已停止" + JobState.getDescByCode(jobDTO.getState()));
                    break;
                case SUCCESS:
                    msg.setTitle("质量作业【" + jobDTO.getName() + "】执行" + JobState.getDescByCode(jobDTO.getState()));
                    break;
                default:
                    continue;
            }
            StringBuffer sb = new StringBuffer();
            sb.append("质量作业名称：" + jobDTO.getName() + "\n").append("作业状态：" + JobState.getDescByCode(jobDTO.getState()) + "\n");
            msg.setContent(sb.toString());
            msg.setSendUser(jobDTO.getCreatedBy());
            msg.setRecvUser(subscribe.getUserId());
            msg.setSourceCode("personal");
            msg.setSourceName("数据质量通知");
            msg.setSendDate(new Date());
            msg.setMsgType("0");
            msg.setSendType("0");
            msg.setSysApplicationId(ThreadContextHelper.getAppId());

            logger.info("待发送系统消息：" + JSONObject.toJSONString(msg));
            this.sysMsgClient.sendmsg(msg);
        }

        return true;
    }

    /**
     * 查询DDS中流程实例状态，更新质量作业状态
     *
     * @param jobList
     */
    private void handleJobState(List<JobDTO> jobList) {
        if (CollectionUtils.isEmpty(jobList)) {
            return;
        }

        List<JobDTO> validJobList = jobList.stream().filter(item -> {
            if (item == null) {
                return false;
            }
            return item.getId() != null;
        }).collect(Collectors.toList());

        List<ProcessInstanceDto> procInsList = this.mapper.getProcInsListByJobId(validJobList);
        if (CollectionUtils.isEmpty(procInsList)) {
            return;
        }
        Map<String, ProcessInstanceDto> procInsMap = procInsList.stream().collect(Collectors.toMap(ProcessInstanceDto::getProcessDefinitionId, Function.identity()));

        List<JobDTO> toUpdateJobList = new ArrayList<>();
        for (JobDTO job : validJobList) {
            ProcessInstanceDto procIns = procInsMap.get(job.getId());
            if (procIns == null || procIns.getState() == null) {
                return;
            }
            if (job.getState() != null && job.getState().equals(procIns.getState())) {
                continue;
            }
            // 启动任务后，JobDTO中state变为“运行中”，但是DDS中流程实例可能为“已提交”状态。此时，前端查询启动后的任务状态，如果不是“运行中”会卡死
            if (ExecutionStatus.SUBMITTED_SUCCESS.getCode() == procIns.getState()) {
                continue;
            }

            job.setState(covertProcInsState2JobState(job.getState(), procIns.getState()));
            toUpdateJobList.add(job);
        }

        // 批量更新作业状态
        if (CollectionUtils.isNotEmpty(toUpdateJobList)) {
            this.mapper.batchUpdateJobState(toUpdateJobList);
        }
    }

    /**
     * 流程实例状态转为质量作业状态
     *
     * @param jobState     状态,0:草稿、1:运行中、5:停止、6:失败、7:成功
     * @param procInsState 流程实例状态:0--提交成功;1--运行中;2--准备暂停;3--暂停;4-准备停止;5--停止;6--失败;7--成功;8-需要容错;9-kill;10-等待线程;11-等待依赖加载
     * @return
     */
    private Integer covertProcInsState2JobState(Integer jobState, Integer procInsState) {
        if (procInsState == null) {
            return jobState;
        }
        if (jobState != null && jobState.equals(procInsState)) {
            return jobState;
        }

        switch (procInsState) {
            case 0:
                return JobState.EDIT.getCode();
            case 1:
                return JobState.RUNNING.getCode();
            case 2:
            case 3:
            case 4:
            case 5:
                return JobState.STOP.getCode();
            case 6:
                return JobState.FAIL.getCode();
            case 7:
                return JobState.SUCCESS.getCode();
            case 8:
            case 9:
            case 10:
            case 11:
                return JobState.FAIL.getCode();
            default:
                throw new BusinessException("流程实例不支持该状态:" + procInsState);
        }
    }

    /**
     * 是否为结构化数据校验
     *
     * @param type
     * @return
     */
    private boolean isStructDataMeasure(Integer type) {
        if (type != null) {
            return type == MeasureType.SINGLE_SOURCE.getCode() || type == MeasureType.MULTI_SOURCE.getCode();
        }

        return false;
    }

    /**
     * map参数转为JobDTO
     *
     * @param map
     * @return
     */
    private JobDTO map2Job(Map<String, Object> map) {
        if (map == null || map.size() <= 0) {
            return null;
        }

        JobDTO jobDTO = new JobDTO();
        jobDTO.setName((String) map.get("name"));
        jobDTO.setMeasureType(MeasureType.FILE.getCode());
        jobDTO.setMeasureTypeName(MeasureType.FILE.getDesc());
        jobDTO.setFileType((Integer) map.get(QaConstant.FILE_TYPE));
        jobDTO.setFileTypeName(FileType.getDescByCode(jobDTO.getFileType()));
        jobDTO.setDataProp((String) map.get("dataProp"));
        jobDTO.setType((String) map.get("type"));
        jobDTO.setTypeName((String) map.get("typeName"));
        jobDTO.setModelCode((String) map.get("modelCode"));
        jobDTO.setAircraftName((String) map.get("aircraftName"));
        jobDTO.setFlightDate((String) map.get("flightDate"));
        jobDTO.setSortie((String) map.get("sortie"));
        jobDTO.setTestSystemName((String) map.get("testSystemName"));
        jobDTO.setSchedulerType(SchedulerType.AUTO.getCode());
        jobDTO.setFileIdList((List<String>) map.get("fileIdList"));
        jobDTO.setResourceIds((String) map.get("resourceIds"));
        jobDTO.setOffsiteDataId((String) map.get("ftdpOffsiteDataId"));
        jobDTO.setName(buildFileJobName(jobDTO));

        return jobDTO;
    }

    /**
     * 生成试飞信息
     *
     * @param jobDTO
     * @return
     */
    private FtdpOffsiteDataDTO covert2OffsiteData(JobDTO jobDTO) {
        if (jobDTO == null) {
            return null;
        }

        FtdpOffsiteDataDTO offsiteData = new FtdpOffsiteDataDTO();
        offsiteData.setId(jobDTO.getOffsiteDataId());
        offsiteData.setFileType(jobDTO.getFileType());
        offsiteData.setFileTypeName(FileType.getDescByCode(jobDTO.getFileType()));
        offsiteData.setType(jobDTO.getType());
        offsiteData.setTypeName(jobDTO.getTypeName());
        offsiteData.setModelCode(jobDTO.getModelCode());
        offsiteData.setAircraftName(jobDTO.getAircraftName());
        offsiteData.setSortie(jobDTO.getSortie());
        offsiteData.setFlightDate(jobDTO.getFlightDate());
        offsiteData.setTestSystemName(jobDTO.getTestSystemName());
        offsiteData.setDataProp(jobDTO.getDataProp());
        offsiteData.setResourceIds(jobDTO.getResourceIds());

        return offsiteData;
    }

    /**
     * 文件校验任务，判断是否自动匹配模板
     *
     * @param jobDTO
     */
    private void autoCreateRule(JobDTO jobDTO) {
        if (isStructDataMeasure(jobDTO.getMeasureType())) {
            return;
        }

        boolean isAutoMatchTemplate = ConfigUtils.getInstance().getBoolean(null, "bdp.qa.task.auto.match.template");
        if (!isAutoMatchTemplate) {
            return;
        }
        this.ruleService.autoCreateRule(jobDTO);
    }

    /**
     * 立即执行文件校验任务
     *
     * @param jobDTO
     * @return
     */
    private ResponseMsg<List<FtdpFileDTO>> executeFileTaskImmediately(JobDTO jobDTO, List<RuleDTO> ruleList) {
        String algoType = null;
        if (FileType.RAW_DATA.getCode() == jobDTO.getFileType()) {
            algoType = "pcm";
        } else {
            algoType = "processData";
        }

        return this.algoStrategySelector.getAlgoByType(algoType).execute(jobDTO, ruleList);
    }

    /**
     * 设置文件校验任务状态（自动执行）
     *
     * @param retCode
     * @param jobDTO
     */
    private void setFileJobState(String retCode, JobDTO jobDTO) {
        jobDTO.setState(retCode.equals(Integer.toString(HttpStatus.OK_200)) ? JobState.SUCCESS.getCode() : JobState.FAIL.getCode());
    }

    /**
     * 设置文件校验任务日志（自动执行）
     *
     * @param result
     * @param execTime
     * @param jobDTO
     */
    private void setFileJobLog(ResponseMsg<List<FtdpFileDTO>> result, String execTime, JobDTO jobDTO) {
        FtdpOffsiteDataDTO offsiteData = JSONObject.parseObject(jobDTO.getAttr(), FtdpOffsiteDataDTO.class);
        if (offsiteData == null) {
            return;
        }

        StringBuffer sb = new StringBuffer();
        sb.append("任务名称：").append(jobDTO.getName() == null ? QaConstant.DOUBLE_HYPHEN : jobDTO.getName()).append("\n");
        sb.append("校验类型：").append(MeasureType.FILE.getDesc()).append("\n");
        sb.append("调度方式：").append(SchedulerType.AUTO.getDesc()).append("\n");
        sb.append("执行时间：").append(execTime).append("\n");
        sb.append("类别：").append(jobDTO.getTypeName() == null ? QaConstant.DOUBLE_HYPHEN : jobDTO.getTypeName()).append("\n");
        sb.append("型号：").append(jobDTO.getModelCode() == null ? QaConstant.DOUBLE_HYPHEN : jobDTO.getModelCode()).append("\n");
        sb.append("机号：").append(jobDTO.getAircraftName() == null ? QaConstant.DOUBLE_HYPHEN : jobDTO.getAircraftName()).append("\n");
        sb.append("飞行日期：").append(jobDTO.getFlightDate() == null ? QaConstant.DOUBLE_HYPHEN : jobDTO.getFlightDate()).append("\n");
        sb.append("架次：").append(jobDTO.getSortie() == null ? QaConstant.DOUBLE_HYPHEN : jobDTO.getSortie()).append("\n");
        sb.append("测试系统：").append(jobDTO.getTestSystemName() == null ? QaConstant.DOUBLE_HYPHEN : jobDTO.getTestSystemName()).append("\n");
        sb.append("工程值文件：").append(getFileNames(result.getResponseBody()) == null ? QaConstant.DOUBLE_HYPHEN : getFileNames(result.getResponseBody())).append("\n");
        if (JobState.SUCCESS.getCode() == jobDTO.getState()) {
            sb.append("执行结果：").append(JobState.SUCCESS.getDesc());
        } else {
            sb.append("执行结果：").append(JobState.FAIL.getDesc()).append("\n");
            sb.append("失败原因：").append(result.getErrorDesc());
        }

        offsiteData.setLog(sb.toString());
        jobDTO.setAttr(JSONObject.toJSONString(offsiteData));
    }

    /**
     * 设置文件校验任务质量报告（自动执行）
     *
     * @param jobDTO
     * @param ruleList
     * @param result
     * @param execTime
     */
    private void setFileJobReport(JobDTO jobDTO, List<RuleDTO> ruleList, ResponseMsg<List<FtdpFileDTO>> result, String execTime) {
        if (!result.getRetCode().equals(Integer.toString(HttpStatus.OK_200))) {
            logger.warn(String.format("任务[%s]执行失败，不需要生成质量报告", jobDTO.getName()));
            return;
        }

        this.jobReportService.buildAutoExecFileJobReport(jobDTO, ruleList, result, execTime);
    }

    /**
     * 自动执行任务执行时间
     *
     * @param startTime
     * @param endTime
     * @return
     */
    private String getExecTime(Date startTime, Date endTime) {
        SimpleDateFormat sdf = new SimpleDateFormat(Constants.YYYY_MM_DD_HH_MM_SS);

        return sdf.format(startTime) + "至" + sdf.format(endTime);
    }

    /**
     * 自动执行任务返回结果
     *
     * @param result
     * @param jobId
     * @return
     */
    private ResponseMsg<String> getResMsg(ResponseMsg<List<FtdpFileDTO>> result, String jobId) {
        ResponseMsg rsp = new ResponseMsg<>();

        rsp.setRetCode(result.getRetCode());
        rsp.setErrorDesc(result.getErrorDesc());
        rsp.setResponseBody(jobId);

        return rsp;
    }

    /**
     * 获取校验文件列表，多个文件使用英文逗号隔开
     *
     * @param fileList
     * @return
     */
    private String getFileNames(List<FtdpFileDTO> fileList) {
        if (CollectionUtils.isNotEmpty(fileList)) {
            List<String> nameList = new ArrayList<>();
            for (FtdpFileDTO file : fileList) {
                nameList.add(file.getName());
            }

            return String.join(QaConstant.COMMA, nameList);
        }

        return null;
    }

    /**
     * 构建文件作业名称（自动执行）
     * 名称格式：日期-测试系统-类别-型号-机号-架次-飞行日期
     *
     * @param jobDTO
     * @return
     */
    private String buildFileJobName(JobDTO jobDTO) {
        if (StringUtils.isNotEmpty(jobDTO.getName())) {
            return jobDTO.getName();
        }

        // 日期
        StringBuilder sb = new StringBuilder();
        sb.append(new SimpleDateFormat("yyyyMMddHHmmss").format(new Date()));
        // 测试系统
        if (StringUtils.isNotEmpty(jobDTO.getTestSystemName())) {
            sb.append("-").append(jobDTO.getTestSystemName());
        }
        // 类别名称
        if (StringUtils.isNotEmpty(jobDTO.getTypeName())) {
            sb.append("-").append(jobDTO.getTypeName());
        }
        // 型号名称
        if (StringUtils.isNotEmpty(jobDTO.getModelCode())) {
            sb.append("-").append(jobDTO.getModelCode());
        }
        // 机号名称
        if (StringUtils.isNotEmpty(jobDTO.getAircraftName())) {
            sb.append("-").append(jobDTO.getAircraftName());
        }
        // 架次
        if (StringUtils.isNotEmpty(jobDTO.getSortie())) {
            sb.append("-").append(jobDTO.getSortie());
        }
        // 飞行日期
        if (StringUtils.isNotEmpty(jobDTO.getFlightDate())) {
            sb.append("-").append(jobDTO.getFlightDate());
        }

        return sb.toString();
    }

    /**
     * 自动执行任务
     *
     * @param schedulerType
     * @return
     */
    private boolean isAutoExecJob(Integer schedulerType) {
        return SchedulerType.AUTO.getCode() == schedulerType;
    }
}
